#include "catalog_manager.h"
#include <fstream>
#include <algorithm>
#include <thread>

CatalogManager::CatalogManager() : logger_(Logger::GetInstance())
{
    loadFromDisk(); // 启动时加载元数据
    logger_.Info("CatalogManager initialized.");
}

bool CatalogManager::createTable(const std::string &table_name, const std::vector<ColumnSchema> &cols, const std::string comment, uint32_t first_page_id)
{
    if (table_map_.count(table_name) > 0)
    {
        logger_.Error("Table '" + table_name + "' already exists.");
        return false;
    }
    // 添加到内存缓存
    TableSchema schema;
    schema.table_name = table_name;
    schema.cols = cols;
    schema.indexes = std::vector<IndexSchema>();
    schema.unique_constraints = std::unordered_map<std::string, std::set<Value>>();
    schema.auto_increment_counters = std::unordered_map<std::string, int64_t>();
    // 初始化自增列计数器
    bool has_primary_key = false;
    for (const auto &col : cols)
    {
        if (col.auto_increment)
        {
            schema.auto_increment_counters[col.name] = 0;
        }
        if (col.primary)
        {
            schema.indexes.push_back(IndexSchema{
                table_name + "_pk", // 按顺序对应index_name
                {col.name},         // 对应columns
                true,               // 对应is_unique
                first_page_id       // 对应root_page_id
            });
            has_primary_key = true;
            schema.primary_key_name = table_name + "_pk";
            schema.unique_constraints[col.name] = std::set<Value>();
            schema.cols[0].unique = true;
            schema.cols[0].nullable = false;
        }
        if (col.unique)
        {
            schema.unique_constraints[col.name] = std::set<Value>();
        }
    }
    if (!has_primary_key)
    {
        schema.cols[0].primary = true; // 强制第一个列为主键
        schema.indexes.push_back(IndexSchema{
            table_name + "_pk",
            {cols[0].name},
            true,
            first_page_id});
        schema.primary_key_name = table_name + "_pk";
        schema.unique_constraints[cols[0].name] = std::set<Value>();
        schema.cols[0].unique = true;
        schema.cols[0].nullable = false;
    }
    schema.primary_page_id = first_page_id;
    schema.comment = comment;
    table_map_[table_name] = schema;
    // 持久化到磁盘
    std::thread save_thread(&CatalogManager::saveToDisk, this);
    save_thread.detach();
    logger_.Info("Created table: " + table_name);
    return true;
}

uint32_t CatalogManager::getFirstPrimaryPageId(const std::string &table_name)
{
    auto it = table_map_.find(table_name);
    if (it == table_map_.end())
        return 0; // 0表示无效页ID
    return it->second.primary_page_id;
}

void CatalogManager::updateIndexRootPageId(const std::string &table_name, const std::string &index_name, uint32_t root_page_id)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto index_it = std::find_if(it->second.indexes.begin(), it->second.indexes.end(),
                                     [&](const IndexSchema &index)
                                     { return index.index_name == index_name; });
        if (index_it != it->second.indexes.end())
        {
            index_it->root_page_id = root_page_id;
            // If this is the primary index, also update the primary_page_id for backward compatibility/convenience.
            if (index_name == it->second.primary_key_name)
            {
                it->second.primary_page_id = root_page_id;
            }
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
        }
    }
}

void CatalogManager::loadFromDisk()
{
    try
    {
        std::ifstream file("catalog.json");
        if (!file.is_open())
        {
            logger_.Info("Catalog file not found, will create new one when needed.");
            return;
        }

        json data;
        file >> data;

        if (data.is_null())
        {
            logger_.Warning("Catalog file is empty");
            return;
        }

        for (auto &[table_name, table_json] : data.items())
        {
            if (!table_json.is_object())
            {
                logger_.Warning("Invalid table format for: " + table_name);
                continue;
            }

            TableSchema schema;
            schema.table_name = table_name;

            // 加载基本表信息
            schema.primary_page_id = table_json.value("primaryPageId", 0);
            schema.comment = table_json.value("comment", "");
            schema.primary_key_name = table_json.value("primaryKeyName", "");

            // 加载列信息
            if (table_json.contains("columns") && table_json["columns"].is_array())
            {
                for (auto &col_json : table_json["columns"])
                {
                    ColumnSchema col;
                    col.name = col_json.value("name", "");
                    col.type = col_json.value("type", "");
                    col.length = col_json.value("length", 0);
                    col.nullable = col_json.value("nullable", false);
                    col.unique = col_json.value("unique", false);
                    col.primary = col_json.value("primary", false);
                    col.auto_increment = col_json.value("autoIncrement", false);
                    col.default_value = col_json.value("defaultValue", std::optional<std::string>());
                    col.comment = col_json.value("comment", "");

                    schema.cols.push_back(col);
                }
            }

            // 加载索引信息
            if (table_json.contains("indexes") && table_json["indexes"].is_array())
            {
                for (auto &index_json : table_json["indexes"])
                {
                    IndexSchema index;
                    index.index_name = index_json.value("indexName", "");
                    index.columns = index_json.value("columns", std::vector<std::string>());
                    index.is_unique = index_json.value("isUnique", false);
                    index.root_page_id = index_json.value("rootPageId", 0);

                    schema.indexes.push_back(index);
                }
            }

            // 加载自增计数器
            if (table_json.contains("autoIncrementCounters") && table_json["autoIncrementCounters"].is_object())
            {
                for (auto &[col_name, counter] : table_json["autoIncrementCounters"].items())
                {
                    schema.auto_increment_counters[col_name] = counter;
                }
            }

            // 加载唯一约束 (*** MODIFIED SECTION ***)
            if (table_json.contains("uniqueConstraints") && table_json["uniqueConstraints"].is_object())
            {
                for (auto &[col_name, values_json] : table_json["uniqueConstraints"].items())
                {
                    if (values_json.is_array())
                    {
                        std::set<Value> value_set;
                        for (const auto &val_json : values_json)
                        {
                            // 使用在 db_common.h 中定义的辅助函数
                            value_set.insert(fromJsonValue(val_json));
                        }
                        schema.unique_constraints[col_name] = value_set;
                    }
                }
            }

            table_map_[table_name] = schema;
        }

        logger_.Info("Successfully loaded catalog with " + std::to_string(table_map_.size()) + " tables.");
    }
    catch (const std::exception &e)
    {
        logger_.Error("Failed to load catalog: " + std::string(e.what()));
        throw;
    }
}

void CatalogManager::saveToDisk()
{
    try
    {
        json data;

        for (auto &[table_name, schema] : table_map_)
        {
            json table_json;

            // 保存基本表信息
            table_json["primaryPageId"] = schema.primary_page_id;
            table_json["comment"] = schema.comment;
            table_json["primaryKeyName"] = schema.primary_key_name;

            // 保存列信息
            json columns_json = json::array();
            for (auto &col : schema.cols)
            {
                json col_json;
                col_json["name"] = col.name;
                col_json["type"] = col.type;
                col_json["length"] = col.length;
                col_json["nullable"] = col.nullable;
                col_json["unique"] = col.unique;
                col_json["primary"] = col.primary;
                col_json["autoIncrement"] = col.auto_increment;
                if (col.default_value.has_value())
                {
                    col_json["defaultValue"] = col.default_value.value();
                }
                col_json["comment"] = col.comment;

                columns_json.push_back(col_json);
            }
            table_json["columns"] = columns_json;

            // 保存索引信息
            json indexes_json = json::array();
            for (auto &index : schema.indexes)
            {
                json index_json;
                index_json["indexName"] = index.index_name;
                index_json["columns"] = index.columns;
                index_json["isUnique"] = index.is_unique;
                index_json["rootPageId"] = index.root_page_id;

                indexes_json.push_back(index_json);
            }
            table_json["indexes"] = indexes_json;

            // 保存自增计数器
            json auto_inc_json;
            for (auto &[col_name, counter] : schema.auto_increment_counters)
            {
                auto_inc_json[col_name] = counter;
            }
            table_json["autoIncrementCounters"] = auto_inc_json;

            // 保存唯一约束
            json unique_constraints_json = json::object();
            for (const auto &[col_name, values_set] : schema.unique_constraints)
            {
                json values_json_array = json::array();
                for (const auto &val : values_set)
                {
                    // 使用在 db_common.h 中定义的辅助函数
                    values_json_array.push_back(toJson(val));
                }
                unique_constraints_json[col_name] = values_json_array;
            }
            table_json["uniqueConstraints"] = unique_constraints_json;

            data[table_name] = table_json;
        }

        std::ofstream file("catalog.json");
        if (!file.is_open())
        {
            throw std::runtime_error("Failed to open catalog file for writing");
        }

        file << std::setw(4) << data << std::endl;
        logger_.Info("Successfully saved " + std::to_string(table_map_.size()) + " tables to catalog");
    }
    catch (const std::exception &e)
    {
        logger_.Error("Failed to save catalog: " + std::string(e.what()));
        throw;
    }
}

TableSchema CatalogManager::getTableSchema(const std::string &table_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        return it->second;
    }
    throw std::runtime_error("Table '" + table_name + "' not found.");
}

void CatalogManager::updateTablePrimaryPageId(const std::string &table_name, uint32_t page_id)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        updateIndexRootPageId(table_name, it->second.primary_key_name, page_id);
    }
}

bool CatalogManager::tableExists(const std::string &table_name)
{
    return table_map_.find(table_name) != table_map_.end();
}

bool CatalogManager::indexExists(const std::string &table_name, const std::string &index_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
        return std::find_if(it->second.indexes.begin(), it->second.indexes.end(), [&](const IndexSchema &index)
                            { return index.index_name == index_name; }) != it->second.indexes.end();
    return false;
}

bool CatalogManager::createIndex(const std::string &table_name, const std::string &index_name, const std::vector<std::string> &cols, uint32_t root_page_id, bool is_unique)
{
    auto it = table_map_.find(table_name);
    if (it == table_map_.end())
    {
        logger_.Error("Table '" + table_name + "' not found.");
        return false;
    }
    if (indexExists(table_name, index_name))
    {
        logger_.Error("Index '" + index_name + "' already exists on table '" + table_name + "'.");
        return false;
    }
    IndexSchema index;
    index.index_name = index_name;
    index.columns = cols;
    index.root_page_id = root_page_id;
    index.is_unique = is_unique;
    it->second.indexes.push_back(index);
    std::thread save_thread(&CatalogManager::saveToDisk, this);
    save_thread.detach();
    logger_.Info("Created index: " + index_name + " on table: " + table_name);
    return true;
}

bool CatalogManager::dropIndex(const std::string &table_name, const std::string &index_name)
{
    auto it = table_map_.find(table_name);
    if (it == table_map_.end())
    {
        logger_.Error("Table '" + table_name + "' not found.");
        return false;
    }
    auto index_it = std::find_if(it->second.indexes.begin(), it->second.indexes.end(), [&](const IndexSchema &index)
                                 { return index.index_name == index_name; });
    if (index_it == it->second.indexes.end())
    {
        logger_.Error("Index '" + index_name + "' not found on table '" + table_name + "'.");
        return false;
    }
    it->second.indexes.erase(index_it);
    std::thread save_thread(&CatalogManager::saveToDisk, this);
    save_thread.detach();
    logger_.Info("Dropped index: " + index_name + " on table: " + table_name);
    return true;
}

bool CatalogManager::uniqueConstraintExists(const std::string &table_name, const std::string &col_name, const Value &val)
{
    auto it = table_map_.find(table_name);
    if (it == table_map_.end())
    {
        logger_.Error("Table '" + table_name + "' not found.");
        return false;
    }
    auto uc_it = it->second.unique_constraints.find(col_name);
    if (uc_it != it->second.unique_constraints.end())
    {
        return uc_it->second.find(val) != uc_it->second.end();
    }
    return false;
}
bool CatalogManager::createUniqueConstraint(const std::string &table_name, const std::string &col_name, const Value &val)
{
    auto it = table_map_.find(table_name);
    if (it == table_map_.end())
    {
        logger_.Error("Table '" + table_name + "' not found.");
        return false;
    }
    auto uc_it = it->second.unique_constraints.find(col_name);
    if (uc_it == it->second.unique_constraints.end())
    {
        logger_.Error("Column '" + col_name + "' not found on table '" + table_name + "'.");
        return false;
    }
    uc_it->second.insert(val);
    std::thread save_thread(&CatalogManager::saveToDisk, this);
    save_thread.detach();
    logger_.Info("Created unique constraint on column: " + col_name + " on table: " + table_name);
    return true;
}
IndexSchema CatalogManager::getIndexSchema(const std::string &table_name, const std::string &index_name) const
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto index_it = std::find_if(it->second.indexes.begin(), it->second.indexes.end(), [&](const IndexSchema &index)
                                     { return index.index_name == index_name; });
        if (index_it != it->second.indexes.end())
        {
            return *index_it;
        }
    }
    throw std::runtime_error("Index '" + index_name + "' not found on table '" + table_name + "'.");
}

void CatalogManager::dropTable(const std::string &table_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        table_map_.erase(it);
        std::thread save_thread(&CatalogManager::saveToDisk, this);
        save_thread.detach();
        logger_.Info("Dropped table: " + table_name);
    }
    else
    {
        throw std::runtime_error("Table '" + table_name + "' not found.");
    }
}

IndexSchema CatalogManager::getIndexSchemaByColumn(const std::string &table_name, const std::string &col_name) const
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto index_it = std::find_if(it->second.indexes.begin(), it->second.indexes.end(), [&](const IndexSchema &index)
                                     { return index.columns.size() == 1 && index.columns[0] == col_name; });
        if (index_it != it->second.indexes.end())
        {
            return *index_it;
        }
    }
    throw std::runtime_error("Index '" + col_name + "' not found on table '" + table_name + "'.");
}

bool CatalogManager::renameTable(const std::string &old_name, const std::string &new_name)
{
    auto it = table_map_.find(old_name);
    if (it != table_map_.end())
    {
        table_map_[new_name] = it->second;
        table_map_.erase(it);
        std::thread save_thread(&CatalogManager::saveToDisk, this);
        save_thread.detach();
        logger_.Info("Renamed table: " + old_name + " to " + new_name);
        return true;
    }
    return false;
}

bool CatalogManager::modifyComment(const std::string &table_name, const std::string &comment)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        it->second.comment = comment;
        std::thread save_thread(&CatalogManager::saveToDisk, this);
        save_thread.detach();
        logger_.Info("Modified comment for table: " + table_name);
        return true;
    }
    return false;
}

bool CatalogManager::addColumns(const std::string &table_name, const std::vector<ColumnSchema> &cols)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        for (auto &col : cols)
        {
            it->second.cols.push_back(col);
        }
        std::thread save_thread(&CatalogManager::saveToDisk, this);
        save_thread.detach();
        logger_.Info("Added columns to table: " + table_name);
        return true;
    }
    return false;
}
bool CatalogManager::dropColumns(const std::string &table_name, const std::vector<std::string> &col_names)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        for (auto &col_name : col_names)
        {
            it->second.cols.erase(std::remove_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                                 { return col.name == col_name; }));
        }
        std::thread save_thread(&CatalogManager::saveToDisk, this);
        save_thread.detach();
        logger_.Info("Dropped columns from table: " + table_name);
        return true;
    }
    return false;
}

bool CatalogManager::modifyColumns(const std::string &table_name, std::unordered_map<std::string, ModifiedColumn> &col_map, std::unordered_map<std::string, int> need_to_auto_increment,
                                   std::unordered_map<std::string, std::set<Value>> need_to_unique,
                                   std::vector<std::string> need_to_drop_auto_increment,
                                   std::vector<std::string> need_to_drop_unique)
{
    auto table_it = table_map_.find(table_name);
    if (table_it != table_map_.end())
    {
        for (auto &col : table_it->second.cols)
        {
            auto it = col_map.find(col.name);
            if (it != col_map.end())
            {
                if (it->second.name.has_value())
                {
                    col.name = it->second.name.value();
                }
                if (it->second.type.has_value())
                {
                    col.type = it->second.type.value();
                }
                if (it->second.length.has_value())
                {
                    col.length = it->second.length.value();
                }
                if (it->second.nullable.has_value())
                {
                    col.nullable = it->second.nullable.value();
                }
                if (it->second.default_value.has_value())
                {
                    col.default_value = it->second.default_value.value();
                }
                if (it->second.comment.has_value())
                {
                    col.comment = it->second.comment.value();
                }
            }
        }
        for (auto &col_name : need_to_drop_auto_increment)
        {
            auto it = std::find_if(table_it->second.cols.begin(), table_it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
            if (it != table_it->second.cols.end())
            {
                it->auto_increment = false;
                table_it->second.auto_increment_counters.erase(col_name);
            }
        }
        for (auto &col_name : need_to_drop_unique)
        {
            auto it = std::find_if(table_it->second.cols.begin(), table_it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
            if (it != table_it->second.cols.end())
            {
                it->unique = false;
                table_it->second.unique_constraints.erase(col_name);
            }
        }
        for (auto &[col_name, value_set] : need_to_unique)
        {
            auto it = std::find_if(table_it->second.cols.begin(), table_it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
            if (it != table_it->second.cols.end())
            {
                it->unique = true;
                table_it->second.unique_constraints[col_name] = value_set;
            }
        }
        for (auto &[col_name, value] : need_to_auto_increment)
        {
            auto it = std::find_if(table_it->second.cols.begin(), table_it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
            if (it != table_it->second.cols.end())
            {
                it->auto_increment = true;
                table_it->second.auto_increment_counters[col_name] = value;
            }
        }
        std::thread save_thread(&CatalogManager::saveToDisk, this);
        save_thread.detach();
        logger_.Info("Modified columns on table: " + table_name);
        return true;
    }
    return false;
}

bool CatalogManager::renameIndex(const std::string &table_name, const std::string &old_name, const std::string &new_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto index_it = std::find_if(it->second.indexes.begin(), it->second.indexes.end(), [&](const IndexSchema &index)
                                     { return index.index_name == old_name; });
        if (index_it != it->second.indexes.end())
        {
            index_it->index_name = new_name;
            if (old_name == it->second.primary_key_name)
            {
                it->second.primary_key_name = new_name;
            }
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
            logger_.Info("Renamed index: " + old_name + " to " + new_name);
            return true;
        }
    }
    return false;
}

bool CatalogManager::addAutoIncrement(const std::string &table_name, const std::string &col_name, int start_value)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            col_it->auto_increment = true;
            it->second.auto_increment_counters[col_name] = start_value;
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
            logger_.Info("Created auto increment on column: " + col_name + " on table: " + table_name);
            return true;
        }
    }
    return false;
}

bool CatalogManager::dropAutoIncrement(const std::string &table_name, const std::string &col_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            col_it->auto_increment = false;
            it->second.auto_increment_counters.erase(col_name);
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
            logger_.Info("Dropped auto increment on column: " + col_name + " on table: " + table_name);
            return true;
        }
    }
    return false;
}

bool CatalogManager::addUniqueConstraint(const std::string &table_name, const std::string &col_name, std::set<Value> &values)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            col_it->unique = true;
            it->second.unique_constraints[col_name] = values;
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
            logger_.Info("Created unique constraint on column: " + col_name + " on table: " + table_name);
            return true;
        }
    }
    return false;
}

bool CatalogManager::dropUniqueConstraint(const std::string &table_name, const std::string &col_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            col_it->unique = false;
            it->second.unique_constraints.erase(col_name);
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
            logger_.Info("Dropped unique constraint on column: " + col_name + " on table: " + table_name);
            return true;
        }
    }
    return false;
}

int64_t CatalogManager::getAutoIncrementCounter(const std::string &table_name, const std::string &col_name)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            int64_t count = ++it->second.auto_increment_counters[col_name];
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
            return count;
        }
    }
    return -1;
}

void CatalogManager::setAutoIncrementCounter(const std::string &table_name, const std::string &col_name, int64_t counter)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            it->second.auto_increment_counters[col_name] = counter;
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
        }
    }
}

void CatalogManager::deleteUniqueConstraint(const std::string &table_name, const std::string &col_name, const Value &val)
{
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            auto &constraints = it->second.unique_constraints[col_name];
            constraints.erase(val);
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
        }
    }
}
void CatalogManager::deleteUniqueConstraint(const std::string &table_name, const std::string &col_name){
    auto it = table_map_.find(table_name);
    if (it != table_map_.end())
    {
        auto col_it = std::find_if(it->second.cols.begin(), it->second.cols.end(), [&](const ColumnSchema &col)
                                   { return col.name == col_name; });
        if (col_it != it->second.cols.end())
        {
            it->second.unique_constraints.erase(col_name);
            std::thread save_thread(&CatalogManager::saveToDisk, this);
            save_thread.detach();
        }
    }
}

